﻿using Google.Protobuf;
using GSE.Network;
using GSE.Rpc;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using GSE.protos;

namespace GSE
{
    public enum LogicPacketType
    {
        RPC = 0,
        EventPush = 1

    }

    public struct ClientConfig
    {
        public String RemoteAdddr;
        public TransportType TransportType;
        public UInt16 EventPushFrequency;
    }

    public class Client
    {
        public const UInt16 DefaultEventPushFreq = 30;
        protected OptionContext optCtx;
        public const UInt16 MaxQueuedCalls = 65535;
        protected ClientConfig cfg;
        /// <summary>
        /// The next transaction identifier.
        /// 当id为0时,表示为对端推送的无需响应的消息, 此时,service name也应为空.
        /// 最大值为40000;
        /// </summary>
        protected Int32 nextTransactionId = 0;
        protected Mutex nextTransactionIdLock = new Mutex();
        protected Transport transport;
        protected TransportType transportType;
        /// <summary>
        /// 已经发出,尚未收到响应的RPC请求结果表. transactionID => Result Message Bytes
        /// </summary>
        protected ConcurrentDictionary<UInt16, BlockingCollection<Response>> pendingRpcCallResults;

        // 服务端发出的请求队列.
        protected BlockingCollection<Request> requestQueue;

        protected CancellationTokenSource netErrCancellationTokenSource;

        protected CancellationTokenSource clientCloseCancellationTokenSource = new CancellationTokenSource();

        protected string remoteAddress;
        protected RequestHandlerContainer reqHandlerContainer;

        protected EventPusher eventPusher;

        protected EventIncomingDispatcher inComingEventDispatcher;


        public Int32 TransportReceiveBufferSize
        {
            get { return transport.ReceiveBufferSize(); }
            set { transport.ReceiveBufferSize(value); }
        }

        public Int32 TransportSendBufferSize
        {
            get { return transport.SendBufferSize(); }
            set { transport.SendBufferSize(value); }
        }

        ~Client()
        {
            transport.Close();
        }

        public void Close()
        {
            clientCloseCancellationTokenSource.Cancel();
            transport.Close();
        }

        public Client(ClientConfig cfg)
        {
            optCtx = new OptionContext
            {
            };

            if (cfg.EventPushFrequency < 1)
            {
                cfg.EventPushFrequency = DefaultEventPushFreq;
            }

            this.cfg = cfg;
            this.transportType = cfg.TransportType;
            this.remoteAddress = cfg.RemoteAdddr;
            switch (transportType)
            {
                case TransportType.UDP:
                    transport = new Network.UDP.UDPTransport(remoteAddress, optCtx);
                    break;

                case TransportType.TCP:
                    transport = new Network.TCP.TCPTransport(remoteAddress, optCtx);
                    break;
                default:
                    throw new UnSupportedTransportTypeException(transportType);
            }

            pendingRpcCallResults = new ConcurrentDictionary<UInt16, BlockingCollection<Response>>();
            requestQueue = new BlockingCollection<Request>(500);
            netErrCancellationTokenSource = new CancellationTokenSource();
            transport.SetNetErrCancellationTokenSource(netErrCancellationTokenSource);
            reqHandlerContainer = new RequestHandlerContainer(optCtx);
            var epCfg = new EventPusherConfig();
            epCfg.DefaultEventPushFrequency = cfg.EventPushFrequency;
            epCfg.SendData = transport.SendData;
            eventPusher = new EventPusher(epCfg);
            inComingEventDispatcher = new EventIncomingDispatcher();
        }

        public void Intialize(bool keeepAlive)
        {
            reqHandlerContainer.Load();
            var ep = eventPusher.GetType();
            // 注册消息/事件订阅的处理逻辑
            reqHandlerContainer.Register("#sub", new RequestHandlerRegister(ep, ep.GetMethod("OnNewSub")));
            // 注册消息/事件注销的处理逻辑
            reqHandlerContainer.Register("#unsub", new RequestHandlerRegister(ep, ep.GetMethod("OnUnSub")));
            transport.Start();
            if (keeepAlive)
            {
                transport.KeepAlive();
            }
            Task.Run(() => loopRecv());
            Task.Run(() => dispatchRequest());
        }

        protected ILogger logger
        {
            get { return Logger.Default; }
        }

        public EventPusher GetEventPusher()
        {
            return eventPusher;
        }
        /// <summary>
        /// 生成本次通讯的事务ID, 客户端生成的ID应当为奇数.
        /// </summary>
        /// <returns>The transaction identifier.</returns>
        protected UInt16 genTransactionId()
        {
            var id = (UInt16)Interlocked.Increment(ref nextTransactionId);
            if (id < 40000)
            {
                if (id % 2 != 1)
                {
                    return genTransactionId();
                }
                return id;
            }
            // 大于40000, 则重置.
            nextTransactionIdLock.WaitOne();
            if (nextTransactionId != 1)
            {
                // 已经被其它线程重置.
                nextTransactionIdLock.ReleaseMutex();
                return genTransactionId();
            }

            nextTransactionId = 1;
            nextTransactionIdLock.ReleaseMutex();
            return id;
        }

        public EventSubscription<MessageType> Sub<MessageType>(protos.SubEvent param, IEventConsumer<MessageType> evtConsumer)
        {
            var resp = new protos.SubEventResponse();
            Call("#sub", param, resp);
            switch (resp.Status)
            {
                case protos.EventRpcStatusTypes.Ok:
                    var evtSub = new EventSubscription<MessageType>(resp, evtConsumer, OnUnSub);
                    inComingEventDispatcher.RegisterSub(evtSub);
                    return evtSub;
                default:
                    throw new Exception($"failed to subsribe event[{param.Name}], server message: {resp.Msg}");
            }
        }

        /// <summary>
        /// 将取消订阅远端的消息时.
        /// </summary>
        /// <param name="esub"></param>
        public void OnUnSub<MessageType>(EventSubscription<MessageType> esub)
        {
            this.UnSub(esub);
            inComingEventDispatcher.UnRegisterSub(esub);
        }

        /// <summary>
        /// 取消订阅远端的消息.
        /// </summary>
        /// <param name="esub"></param>
        /// <returns></returns>
        public bool UnSub<MessageType>(EventSubscription<MessageType> esub)
        {
            var req = new protos.CancelSubEvent();
            req.SeriesID = esub.SeriesID;
            var resp = new protos.CancelSubEventResponse();
            Call("unsub", req, resp);
            switch (resp.Status)
            {
                case protos.EventRpcStatusTypes.Ok:
                    return true;
                default:
                    throw new Exception($"failed to un-subsribe event[{esub.SeriesID}], server message: {resp.Msg}");
            }
        }

        /// <summary>
        ///  请求一个远端服务,并等待返回数据.
        /// </summary>
        /// <param name="serviceName"></param>
        /// <param name="param"></param>
        /// <param name="resultType">期望返回的数据类型对象</param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        /// <exception cref="RpcCallTimeoutException"></exception>
        /// <exception cref="RpcSeviceNotExistsException"></exception>
        /// <exception cref="RpcServiceParamException"></exception>
        /// <exception cref="RpcServiceProcessException"></exception>
        /// <exception cref="RpcServiceUnknownException"></exception>
        public IMessage Call(string serviceName, IMessage param, IMessage resultType)
        {
            var req = new Request(serviceName, genTransactionId(), param);
            var result = new BlockingCollection<Response>(1);
            var ok = pendingRpcCallResults.TryAdd(req.TransactionID, result);
            if (!ok)
            {
                throw new Exception("rpc call failed, there may be duplicated pending transaction id.");
            }
            try
            {
                transport.SendData(req.Encode());
            }
            catch (Exception ex)
            {
                pendingRpcCallResults.TryRemove(req.TransactionID, out result);
                throw ex;
            }

            Response rsp;
            try
            {
                ok = result.TryTake(out rsp, 5000, netErrCancellationTokenSource.Token);
            }
            catch (OperationCanceledException)
            {
                throw new RpcServiceConnectionException(remoteAddress);
            }

            if (!ok)
            {
                pendingRpcCallResults.TryRemove(req.TransactionID, out result);
                throw new RpcCallTimeoutException(5000);
            }
            pendingRpcCallResults.TryRemove(req.TransactionID, out result);
            logger.Debug($"Got RPC call response: {rsp}");
            switch (rsp.GetStatus())
            {
                case ResponseStatus.Ok:
                    break;
                case ResponseStatus.ServiceNotExists:
                    throw new RpcSeviceNotExistsException(serviceName);
                case ResponseStatus.ParamIncorrect:
                    throw new RpcServiceParamException();
                case ResponseStatus.ServiceProcessError:
                    throw new RpcServiceProcessException();
                default:
                    throw new RpcServiceUnknownException(rsp.GetStatus());
            }

            resultType.MergeFrom(rsp.Data);

            return resultType;
        }

        protected void loopRecv()
        {
            while (true)
            {
                try
                {
                    RpcObject rpc;
                    if (transportType == TransportType.UDP)
                    {
                        rpc = recvUDP();
                    }
                    else if (transportType == TransportType.TCP)
                    {
                        rpc = recvTCP();
                    }
                    else
                    {
                        throw new UnSupportedTransportTypeException(transportType);
                    }

                    if (rpc is Request)
                    {
                        requestQueue.Add((Request)rpc);
                    }
                    else if (rpc is Push)
                    {
                        inComingEventDispatcher.ReceivePush((Push)rpc);
                    }
                    else
                    {
                        var ok = pendingRpcCallResults.TryGetValue(rpc.TransactionID, out var rspc);
                        if (!ok)
                        {
                            logger.Warn($"Transaction({rpc.TransactionID}) is not in the RPC call catalog, response dropped!");
                            continue;
                        }
                        rspc.Add((Response)rpc);
                    }
                }
                catch (InvalidProtocolDataException ex)
                {
                    logger.Warn($"recvRpc exception: {ex.Message}");
                }
                catch (Exception ex)
                {
                    logger.Warn($"recv stopped, recvRpc exception: {ex.Message}, {ex.StackTrace}");
                    return;
                }
            }
        }

        protected void dispatchRequest()
        {
            while (true)
            {
                try
                {
                    var req = requestQueue.Take(clientCloseCancellationTokenSource.Token);
                    logger.Debug($"dispatching request from server: {req.GetServiceName()} [{req.TransactionID}]");
                    Task.Run(() =>
                    {
                        logger.Debug($"start processing request: {req.GetServiceName()} [{req.TransactionID}]");
                        var response = reqHandlerContainer.Handle(req);
                        logger.Debug($"request processed, sending response: {req.GetServiceName()} [{req.TransactionID}]");
                        transport.SendData(response.Encode());
                    },
                    clientCloseCancellationTokenSource.Token
                    );
                }
                catch (OperationCanceledException)
                {
                    logger.Debug("client is closing, exit dispatch request task.");
                    return;
                }
                catch (Exception ex)
                {
                    logger.Warn($"Un-expected exctpion on request {ex.Message}");
                }

            }
        }

        public bool IsValidLogicPacketType(LogicPacketType t)
        {
            switch (t)
            {
                case LogicPacketType.RPC:
                case LogicPacketType.EventPush:
                    return true;
                default:
                    return false;
            }
        }

        protected RpcObject recvTCP()
        {
            var verB = new byte[1];
            var n = transport.ReadData(verB);
            if (n != 1)
            {
                throw new InvalidProtocolDataException("failed to read protocol version no.");
            }

            var protoVer = (byte)(verB[0] >> 3);
            if (protoVer != RpcObject.protoVerNo)
            {
                throw new InvalidProtocolDataException($"expects protocol version \"{RpcObject.protoVerNo}\", but got \"{protoVer}\"");
            }

            // 屏蔽左边五位,计算action type
            var actionType = (LogicPacketType)(verB[0] & (byte)7);
            if (!IsValidLogicPacketType(actionType))
            {
                throw new InvalidProtocolDataException($"Invalid action type {actionType}");
            }

            var pb = new byte[4];
            n = transport.ReadData(pb);
            if (n != pb.Length)
            {
                throw new InvalidProtocolDataException("failed to read packet length.");
            }
            Array.Reverse(pb);
            var packetLength = BitConverter.ToUInt32(pb, 0);
            var packetData = new byte[packetLength];
            n = transport.ReadData(packetData);
            if (n != packetLength)
            {
                throw new InvalidProtocolDataException($"failed to read packet data, expected {packetLength}, but got {n}");
            }
            if (actionType == LogicPacketType.EventPush)
            {
                return new Push(packetData);
            }

            var offset = 0;
            var transactionIdB = new byte[2];
            Array.Copy(packetData, offset, transactionIdB, 0, transactionIdB.Length);
            Array.Reverse(transactionIdB);
            var transactionId = BitConverter.ToUInt16(transactionIdB, 0);
            offset += transactionIdB.Length;
            bool isResponse = false;
            isResponse = transactionId % 2 == 1;
            if (isResponse)
            {
                var responseStatus = (ResponseStatus)packetData[offset];
                offset += 1;
                var messageData = new byte[packetLength - 3];
                Array.Copy(packetData, offset, messageData, 0, messageData.Length);
                logger.Debug($"received response from server for tansaction [{transactionId}]");
                return new Response(responseStatus, transactionId, messageData);
            }

            // 否则为服务端发起的请求.
            var serviceNameLength = packetData[offset];
            offset += 1;
            var serviceNameB = new byte[serviceNameLength];
            Array.Copy(packetData, offset, serviceNameB, 0, serviceNameLength);
            var serviceName = Encoding.UTF8.GetString(serviceNameB);
            offset += serviceNameLength;
            var paramB = new byte[packetData.Length - offset];
            Array.Copy(packetData, offset, paramB, 0, paramB.Length);
            logger.Debug($"received request from server: {serviceName} [{transactionId}]");
            return new Request(serviceName, transactionId, paramB);
        }


        // 只适合UDP协议.
        protected RpcObject recvUDP()
        {
            var rawData = transport.ReadData();
            if (rawData.Length < 5)
            {
                throw new InvalidProtocolDataException($"data length {rawData.Length} is less than 5.");
            }
            var offset = 0;
            var protoVer = (byte)(rawData[offset] >> 3);
            if (protoVer != RpcObject.protoVerNo)
            {
                throw new InvalidProtocolDataException($"expects protocol version \"{RpcObject.protoVerNo}\", but got \"{protoVer}\"");
            }

            // 屏蔽左边五位,计算action type
            var actionType = (LogicPacketType)(rawData[offset] & (byte)7);
            if (!IsValidLogicPacketType(actionType))
            {
                throw new InvalidProtocolDataException($"Invalid action type {actionType}");
            }

            offset += 1;
            var pb = new byte[4];
            Array.Copy(rawData, offset, pb, 0, pb.Length);
            Array.Reverse(pb);
            var packetLength = BitConverter.ToUInt32(pb, 0);
            if (rawData.Length - 5 != packetLength)
            {
                throw new InvalidProtocolDataException($"data length is expected to be {packetLength}, but got data with length {rawData.Length - 5}");
            }
            offset += pb.Length;

            if (actionType == LogicPacketType.EventPush)
            {
                var pushParamB = new byte[packetLength];
                Array.Copy(rawData, offset, pushParamB, 0, packetLength);
                var rpcPush = new Push(pushParamB);
                return rpcPush;
            }

            var transactionIdB = new byte[2];
            Array.Copy(rawData, offset, transactionIdB, 0, transactionIdB.Length);
            Array.Reverse(transactionIdB);
            var transactionId = BitConverter.ToUInt16(transactionIdB, 0);
            offset += transactionIdB.Length;
            bool isResponse = false;
            isResponse = transactionId % 2 == 1;
            if (isResponse)
            {
                var responseStatus = (ResponseStatus)rawData[offset];
                offset += 1;
                var messageData = new byte[packetLength - 3];
                Array.Copy(rawData, offset, messageData, 0, messageData.Length);
                logger.Debug($"received response from server for tansaction [{transactionId}]");
                return new Response(responseStatus, transactionId, messageData);
            }
            // 否则为服务端发起的请求.
            var serviceNameLength = rawData[offset];
            offset += 1;
            var serviceNameB = new byte[serviceNameLength];
            Array.Copy(rawData, offset, serviceNameB, 0, serviceNameLength);
            var serviceName = Encoding.UTF8.GetString(serviceNameB);
            offset += serviceNameLength;
            var paramB = new byte[rawData.Length - offset];
            Array.Copy(rawData, offset, paramB, 0, paramB.Length);
            logger.Debug($"received request from server: {serviceName} [{transactionId}]");
            return new Request(serviceName, transactionId, paramB);
        }

        /// <summary>
        /// 发布一个消息/事件.
        /// 需先调用EventPusher的RegisterSubscribableEvent方法对此类事件的推送进行配置初始化.
        /// </summary>
        /// <param name="name"></param>
        /// <param name="evt"></param>
        public void PubEvent(string name, IMessage evt)
        {
            eventPusher.Pub(name, evt);
        }
        /// <summary>
        /// 异步地请求一个服务端方法,并立即返回. 服务端返回后调用callback进行异步处理.
        /// </summary>
        //public async void Call(string serviceName, IMessage param, IMessage resultType, Func<IMessage> callbck){
        //    resultType = await Call(serviceName, param, resultType);
        //}
    }
}